package netty.rpc.framework.core.netty.consumer;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import netty.rpc.framework.core.netty.factory.SingletonFactory;
import netty.rpc.framework.core.netty.transport.protocol.RPCMessage;
import netty.rpc.framework.core.netty.transport.protocol.RPCRequest;
import netty.rpc.framework.core.netty.transport.protocol.RPCResponse;
import netty.rpc.framework.core.netty.registry.ServiceDiscovery;
import netty.rpc.framework.core.netty.registry.impl.ZKServiceDiscoveryImpl;

import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;

/**
 * @author guiyiqiao
 * @date 2020/12/11
 */
@Slf4j
public class ClientTransport {

    private ChannelProvider channelProvider;

    private ServiceDiscovery serviceDiscovery;

    private UncompletedRequest uncompletedRequest;

    public ClientTransport() {
        this.channelProvider = SingletonFactory.getInstance(ChannelProvider.class);
        this.serviceDiscovery = SingletonFactory.getInstance(ZKServiceDiscoveryImpl.class);
        this.uncompletedRequest = SingletonFactory.getInstance(UncompletedRequest.class);
    }

    /**
     * 客户端发送RPCRequest实现类
     * 1.从服务注册中心获取服务地址
     * 2.获取服务地址数据传输通道
     * 3.封装请求到待完成请求任务列表并发送
     * @param rpcRequest
     * @return
     */

    public CompletableFuture<RPCResponse> sendRPCRequest(@NonNull RPCRequest rpcRequest){

        InetSocketAddress address = serviceDiscovery.lookupService(rpcRequest.getInterfaceName());

        Channel channel = channelProvider.getChannel(address);
        //do service
        RPCMessage message = RPCMessage.builder()
                .data(rpcRequest)
                .build();
        CompletableFuture<RPCResponse> resultFuture = new CompletableFuture<>();

        ChannelFuture channelFuture = channel.writeAndFlush(message).addListener(
                (ChannelFutureListener) future -> {
                            if (future.isSuccess()) {
                                log.info("client send message: [{}]", message);
                            } else {
                                future.channel().close();
                                resultFuture.completeExceptionally(future.cause());
                                log.error("Send failed:", future.cause());
                            }
                        });
        uncompletedRequest.addFuture(rpcRequest.getRequestId(),resultFuture);

        return resultFuture;
    }
}
